package cn.itcast.streaming.task;

import cn.itcast.streaming.bean.ItcastDataObj;
import cn.itcast.streaming.bean.TripModel;
import cn.itcast.streaming.function.watermark.TripDriveWaterMark;
import cn.itcast.streaming.function.window.TripDriveSimpleWindowFunction;
import cn.itcast.streaming.function.window.TripDriveWindowFunction;
import cn.itcast.streaming.sink.TripDriveSampleToHBaseSink;
import cn.itcast.streaming.sink.TripDriveToHbaseSink;
import cn.itcast.streaming.utils.JsonParseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class TripDriveTask extends BaseTask{
    public static void main(String[] args) {
        /**
         * 实现步骤
         *  1）初始化流处理环境
         *  2）接入kafka的数据源并读取
         *  3）将消费到的json数据转化为javabean
         *  4)过滤出驾驶行程相关的数据
         *  5）对驾驶行程数据应用水印
         *  6）对添加水印的数据进行分组操作，会话窗口
         *  7）驾驶行程采样数据的业务开发
         *    7.1）对数据应用自定义窗口的业务逻辑处理
         *    7.2）将自定义驾驶行程采样业务处理后的数据写入hbasezhong
         *  8）驾驶行程业务逻辑开发
         *    8.1）驾驶行程数据应用自定义窗口的业务逻辑处理
         *    8.2）将自定义驾驶行程业务处理后的数据写入hbase中
         *  9）地交作业
         */

        //todo 1）初始化流处理环境
        StreamExecutionEnvironment env = getEnv("TripDriveTask");
        env.setParallelism(1);
        //todo 2）接入kafka的数据源并读取
        DataStream<String> kafkaStream = createKafkaStream(SimpleStringSchema.class);
        //kafkaStream.print("原始数据>>>");
        //todo 3）将消费到的json数据转化为javabean
        SingleOutputStreamOperator<ItcastDataObj> itcastDataObjectStream = kafkaStream
                .filter(obj -> StringUtils.isNotEmpty(obj)).map(JsonParseUtil::parseJsonToObject);
        //itcastDataObjectStream.print("javabean>>>");
        //todo 4)过滤出驾驶行程相关的数据(状态为2或3的数据为驾驶行程数据）
        SingleOutputStreamOperator<ItcastDataObj> tripDriveDataStream = itcastDataObjectStream
                .filter(itcastDataObj -> itcastDataObj.getChargeStatus() == 2
                        || itcastDataObj.getChargeStatus() == 3);
        //tripDriveDataStream.printToErr("驾驶行程数据>>>>");
        //todo 5）对驾驶行程数据应用水印
        SingleOutputStreamOperator<ItcastDataObj> tripDriveWaterMarkDataStream =
                tripDriveDataStream.assignTimestampsAndWatermarks(new TripDriveWaterMark());
        //todo 6）对添加水印的数据进行分组操作，会话窗口
        WindowedStream<ItcastDataObj, String, TimeWindow> tripDriveWindowStream = tripDriveWaterMarkDataStream
                .keyBy(ItcastDataObj::getVin)
                .window(EventTimeSessionWindows.withGap(Time.minutes(15)));
        //todo 7）驾驶行程采样数据的业务开发
        //todo 7.1）对数据应用自定义窗口的业务逻辑处理
        SingleOutputStreamOperator<String[]> tripDriveSampleDataStream = tripDriveWindowStream
                .apply(new TripDriveSimpleWindowFunction());
        tripDriveSampleDataStream.printToErr("驾驶行程采样数据>>>");
        //todo 7.2）将自定义驾驶行程采样业务处理后的数据写入hbase中
        tripDriveSampleDataStream.addSink(new TripDriveSampleToHBaseSink("TRIPDB:trip_sample"));
        //todo 8）驾驶行程业务逻辑开发
        //todo 8.1）驾驶行程数据应用自定义窗口的业务逻辑处理
        SingleOutputStreamOperator<TripModel> tripDriveWindowDataStream = tripDriveWindowStream.apply(new TripDriveWindowFunction());
        tripDriveWindowDataStream.printToErr("驾驶行程数据分析结果>>>>");
        //todo 8.2）将自定义驾驶行程业务处理后的数据写入hbase中
        tripDriveWindowDataStream.addSink(new TripDriveToHbaseSink("TRIPDB:trip_division"));
        //todo 9）地交作业
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
